package org.apache.rocketmq.client.java.impl.producer;


import apache.rocketmq.v2.HeartbeatRequest;
import apache.rocketmq.v2.NotifyClientTerminationRequest;
import apache.rocketmq.v2.SendMessageRequest;
import apache.rocketmq.v2.SendMessageResponse;
import com.google.common.base.Stopwatch;
import com.google.common.math.IntMath;
import com.google.common.util.concurrent.*;
import io.grpc.Metadata;
import org.apache.rocketmq.client.java.clientapis.ClientConfiguration;
import org.apache.rocketmq.client.java.clientapis.ClientException;
import org.apache.rocketmq.client.java.clientapis.message.Message;
import org.apache.rocketmq.client.java.clientapis.message.MessageId;
import org.apache.rocketmq.client.java.clientapis.producer.*;
import org.apache.rocketmq.client.java.exception.InternalErrorException;
import org.apache.rocketmq.client.java.exception.TooManyRequestsException;
import org.apache.rocketmq.client.java.hook.MessageHookPoints;
import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
import org.apache.rocketmq.client.java.impl.ClientImpl;
import org.apache.rocketmq.client.java.impl.ClientSettings;
import org.apache.rocketmq.client.java.message.MessageCommon;
import org.apache.rocketmq.client.java.message.MessageType;
import org.apache.rocketmq.client.java.message.PublishingMessageImpl;
import org.apache.rocketmq.client.java.retry.ExponentialBackoffRetryPolicy;
import org.apache.rocketmq.client.java.retry.RetryPolicy;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.apache.rocketmq.client.java.route.MessageQueueImpl;
import org.apache.rocketmq.client.java.rpc.RpcInvocation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static com.google.common.base.Preconditions.checkNotNull;



/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/10
 * @方法描述：生产者客户端实现类，这个类的对象就是生产者客户端
 */
@SuppressWarnings({"UnstableApiUsage", "NullableProblems"})
class ProducerImpl extends ClientImpl implements Producer {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerImpl.class);

    //生产者客户端配置信息对象
    protected final ProducerSettings producerSettings;

    //事务消息检查器，这个检查器在第九版本代码中还用不到
    private final TransactionChecker checker;
    //存储主题和对应的负载均衡器的成员变量
    private final ConcurrentMap<String/* topic */, PublishingLoadBalancer> publishingRouteDataCache;
    //构造方法
    ProducerImpl(ClientConfiguration clientConfiguration, Set<String> topics, int maxAttempts,
                 TransactionChecker checker) {
        //调用父类构造方法
        super(clientConfiguration, topics);
        //创建重试策略对象
        ExponentialBackoffRetryPolicy retryPolicy = ExponentialBackoffRetryPolicy.immediatelyRetryPolicy(maxAttempts);
        //创建生辰这客户端配置信息对象
        this.producerSettings = new ProducerSettings(clientId, endpoints, retryPolicy,
                clientConfiguration.getRequestTimeout(), topics);
        //给事务消息检查器赋值
        this.checker = checker;
        this.publishingRouteDataCache = new ConcurrentHashMap<>();
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：启动生产者客户端的方法
     */
    @Override
    protected void startUp() throws Exception {
        try {
            LOGGER.info("Begin to start the rocketmq producer, clientId={}", clientId);
            //调用父类方法启动客户端
            super.startUp();
            LOGGER.info("The rocketmq producer starts successfully, clientId={}", clientId);
        } catch (Throwable t) {
            LOGGER.error("Failed to start the rocketmq producer, try to shutdown it, clientId={}", clientId, t);
            //发生异常则关闭客户端
            shutDown();
            throw t;
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：关闭客户端的方法
     */
    @Override
    protected void shutDown() throws InterruptedException {
        LOGGER.info("Begin to shutdown the rocketmq producer, clientId={}", clientId);
        super.shutDown();
        LOGGER.info("Shutdown the rocketmq producer successfully, clientId={}", clientId);
    }


    @Override
    public ClientSettings getClientSettings() {
        return producerSettings;
    }

    //包装客户端TerminationRequest请求的方法
    @Override
    public NotifyClientTerminationRequest wrapNotifyClientTerminationRequest() {
        return NotifyClientTerminationRequest.newBuilder().build();
    }

    //包装客户端心跳请求的方法
    @Override
    public HeartbeatRequest wrapHeartbeatRequest() {
        return HeartbeatRequest.newBuilder().build();
    }


    //生产者客户端向服务端发送消息的方法，以下方法在第九版本代码中都用不到
    //等后面版本在为大家真正实现
    @Override
    public SendReceipt send(Message message) throws ClientException {
        final ListenableFuture<SendReceipt> future = Futures.transform(
                //在这里执行了发送消息的操作，第二个参数表示当前消息是否为事务消息，false表示不是
                //该方法会返回一个future对象，当这个future对象被设置为完成时，才会执行sendReceipts -> sendReceipts.iterator().next()方法
                send0(Collections.singletonList(message),false),
                sendReceipts -> sendReceipts.iterator().next(),
                MoreExecutors.directExecutor()
                );
        return handleClientFuture( future);
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：生产者客户端内部发布消息到服务端的方法
     */
    private ListenableFuture<List<SendReceiptImpl>> send0(List<Message> messages, boolean txEnabled) {
        //创建一个可设置的Future对象，用于返回发送消息的结果
        SettableFuture<List<SendReceiptImpl>> future = SettableFuture.create();
        //发送消息之前先检测客户端是否处于正在运行状态
        if (!this.isRunning()) {
            //如果状态异常则向future设置异常完成结果
            final IllegalStateException e = new IllegalStateException("Producer is not running now");
            future.setException(e);
            LOGGER.error("Unable to send message because producer is not running, state={}, clientId={}", this.state(), clientId);
            return future;
        }

        //创建发布消息列表，这个PublishingMessageImpl就是真正发送的消息对象
        List<PublishingMessageImpl> pubMessages = new ArrayList<>();
        //便利要发送的消息列表
        for (Message message : messages) {
            try {
                //把消息转换成发布消息对象
                final PublishingMessageImpl pubMessage = new PublishingMessageImpl(message, producerSettings, txEnabled);
                //把发布消息对象添加到发布消息列表中
                pubMessages.add(pubMessage);
            } catch (Throwable t) {
                LOGGER.error("Failed to refine message to send, clientId={}, message={}", clientId, message, t);
                future.setException(t);
                return future;
            }
        }

        //得到要发送的消息的所有主题集合，这里的Set集合中只有一个元素，因为再一次发送消息的操作中，不可能把消息发送到不同的主题下
        //这些要发送的消息的主题肯定是相同的
        final Set<String> topics = pubMessages.stream().map(Message::getTopic).collect(Collectors.toSet());
        //判断集合的大小，如果大于1则说明集合中存在多个主题，此时不能继续发送消息，返回参数异常对象
        if (1 < topics.size()) {
            final IllegalArgumentException e = new IllegalArgumentException("Messages to send have different topics");
            future.setException(e);
            LOGGER.error("Messages to be sent have different topics, no need to proceed, topic(s)={}, clientId={}", topics, clientId);
            return future;
        }

        //获取集合中的那一个主题，这时候就得到了所有消息要发送的主题信息
        final String topic = topics.iterator().next();

        //在这里得到所有消息的类型，这里的set集合中肯定也只有一个元素，因为一个主题下不可能发送多个消息类型
        final Set<MessageType> messageTypes = pubMessages.stream()
                .map(PublishingMessageImpl::getMessageType)
                .collect(Collectors.toSet());

        //判断集合的大小，如果大于1则说明集合中存在多个消息类型，此时不能继续发送消息，返回参数异常对象
        if (1 < messageTypes.size()) {
            final IllegalArgumentException e = new IllegalArgumentException("Messages to send have different types, "
                    + "please check");
            future.setException(e);
            LOGGER.error("Messages to be sent have different message types, no need to proceed, topic={}, messageType"
                    + "(s)={}, clientId={}", topic, messageTypes, clientId, e);
            return future;
        }

        //获取集合中的那一个主题，这时候就得到了所有消息的类型
        final MessageType messageType = messageTypes.iterator().next();
        final String messageGroup;

        //判断要发送的消息是否为顺序消息
        if (MessageType.FIFO.equals(messageType)) {
            //如果这一批消息是顺序消息，那么这些消息肯定都被设置了相同的消息组
            //所以接下来的操作就是把所有消息的消息组收集到一个set集合中
            //如果不出意外，那么这个set集合中的元素也应该是1个
            final Set<String> messageGroups = pubMessages.stream()
                    .map(PublishingMessageImpl::getMessageGroup)
                    .filter(Optional::isPresent)
                    .map(Optional::get)
                    .collect(Collectors.toSet());

            //判断集合的大小，如果大于1则说明集合中存在多个消息组，此时不能继续发送消息，返回参数异常对象
            if (1 < messageGroups.size()) {
                final IllegalArgumentException e = new IllegalArgumentException("FIFO messages to send have different "
                        + "message groups, messageGroups=" + messageGroups);
                future.setException(e);
                LOGGER.error("FIFO messages to be sent have different message groups, no need to proceed, topic={}, "
                        + "messageGroups={}, clientId={}", topic, messageGroups, clientId, e);
                return future;
            }
            //在这里得到了顺序消息的分组信息
            messageGroup = messageGroups.iterator().next();
        } else {
            //如果不是顺序消息，那就没必要获取消息组信息
            messageGroup = null;
        }

        //把要发送的消息主题添加到主题集合中
        this.topics.add(topic);

        //得到该主题下的负载均衡器
        final ListenableFuture<PublishingLoadBalancer> routeFuture = getPublishingTopicRouteResult(topic);
        //将得到的负载均衡器的结果进行转换，所谓转换其实就是从负载均衡器中获取主题路由数据，然后执行发送消息的操作
        return Futures.transformAsync(routeFuture, result -> {
            //判断消息是不是顺序消息，如果messageGroup不为空，则意味着当前要发送的消息为顺序消息
            final List<MessageQueueImpl> candidates = null == messageGroup ?
                    //如果messageGroup为空，则意味着是普通消息，那么就可以从负载均衡器中获取多个候选队列用于发送消息了，发送失败则可以重试
                    takeMessageQueues(result) :
                    //如果是顺序消息，那就从负载均衡器中获取对应的固定队列
                    Collections.singletonList(result.takeMessageQueueByMessageGroup(messageGroup));
            final SettableFuture<List<SendReceiptImpl>> future0 = SettableFuture.create();

            //在这里执行真正发送消息的操作
            send0(future0, topic, messageType, candidates, pubMessages, 1);
            return future0;
        }, MoreExecutors.directExecutor());
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：真正把消息发送给服务器的方法
     */
    private void send0(SettableFuture<List<SendReceiptImpl>> future, String topic, MessageType messageType,
                       final List<MessageQueueImpl> candidates, final List<PublishingMessageImpl> messages, final int attempt) {
        Metadata metadata;
        try {
            //得到客户端签名信息
            metadata = sign();
        } catch (Throwable t) {
            future.setException(t);
            return;
        }
        //使用轮询的方法从候选消息队列中获取一个消息队列
        final MessageQueueImpl messageQueue = candidates.get(IntMath.mod(attempt - 1, candidates.size()));

        //得到当前得到了的消息队列接收的消息类型
        final List<MessageType> acceptMessageTypes = messageQueue.getAcceptMessageTypes();
        //判断程序是否启用了消息类型验证，然后验证队列是否支持要发送的消息类型
        //不匹配则抛出类型不匹配异常
        if (producerSettings.isValidateMessageType() && !acceptMessageTypes.contains(messageType)) {
            final IllegalArgumentException e = new IllegalArgumentException("Current message type not match with "
                    + "topic accept message types, topic=" + topic + ", actualMessageType=" + messageType + ", "
                    + "acceptMessageTypes=" + acceptMessageTypes);
            future.setException(e);
            return;
        }
        //得到消息队列所属Broker节点的网络地址
        final Endpoints endpoints = messageQueue.getBroker().getEndpoints();
        //在这里把要发送的消息包装到SendMessageRequest请求对象中
        final SendMessageRequest request = wrapSendMessageRequest(messages);

        //使用客户端管理器把请求发送给服务端，这里就执行了真正发送消息的操作，这里返回了一个responseFuture对象
        //当客户端接收到服务端响应后，这个responseFuture就会被设置为完成状态
        final ListenableFuture<RpcInvocation<SendMessageResponse>> responseFuture =
                clientManager.sendMessage(endpoints, metadata, request, clientConfiguration.getRequestTimeout());

        //当responseFuture被设置为完成状态，也就意味着responseFuture得到了服务端回复的响应，这个时候transformAsync方法中的回调方法就会执行
        //把服务端回复的响应转换为SendReceiptImpl对象
        final ListenableFuture<List<SendReceiptImpl>> attemptFuture = Futures.transformAsync(responseFuture,
                invocation -> Futures.immediateFuture(SendReceiptImpl.processSendMessageResponseInvocation(messageQueue,
                        invocation)),
                MoreExecutors.directExecutor());

        //得到客户端发送请求的最大重试次数
        final int maxAttempts = this.getRetryPolicy().getMaxAttempts();
        //得到消息的公共信息
        final List<MessageCommon> messageCommons =
                messages.stream().map(PublishingMessageImpl::getMessageCommon).collect(Collectors.toList());
        final Stopwatch stopwatch = Stopwatch.createStarted();
        //执行消息发送前的回调方法
        //虽然上面执行了发送消息的操作，但操作是异步执行的，并不会早于当前的doBefore方法
        doBefore(MessageHookPoints.SEND, messageCommons);

        //在这里又定义了一个回调方法，会根据attemptFuture结果判断执行相应操作
        Futures.addCallback(attemptFuture, new FutureCallback<List<SendReceiptImpl>>() {

            //当attemptFuture被设置为成功状态时，就会执行成功回调方法
            @Override
            public void onSuccess(List<SendReceiptImpl> sendReceipts) {
                //得到发送消息的耗时
                final Duration duration = stopwatch.elapsed();
                //执行消息发送后的回调方法
                doAfter(MessageHookPoints.SEND, messageCommons, duration, MessageHookPointsStatus.OK);
                //这里有一个比较重要的操作，那就是判断发送消息成功后收到的响应回执对象的数量是否和消息发送的数量一致
                //一条消息应该对应一个响应回执对象
                if (sendReceipts.size() != messages.size()) {
                    //如果数量不一致则创建一个内部异常对象
                    final InternalErrorException e = new InternalErrorException("[Bug] due to an"
                            + " unknown reason from remote, received send receipt's quantity " + sendReceipts.size()
                            + " is not equal to sent message's quantity " + messages.size());
                    future.setException(e);
                    return;
                }
                //这里把响应回执对象集合设置到future对象中
                //因为消息发送成功了，可以给外层方法要使用future设置完成状态了
                future.set(sendReceipts);
                //判断当前是第几次重试，如果attempt大于1，意味着之前消息发送失败了，本次成功是消息重试的结果
                //现在大家还不知道在哪里发生了失败重试，接着往下看就知道了
                if (1 < attempt) {
                    //创建一个集合用于收集所有消息的Id
                    List<MessageId> messageIds = new ArrayList<>();
                    //遍历消息回执对象，收集消息ID
                    for (SendReceipt receipt : sendReceipts) {
                        messageIds.add(receipt.getMessageId());
                    }
                    //记录消息发送重试成功的日志
                    LOGGER.info("Resend message successfully, topic={}, messageId(s)={}, maxAttempts={}, "
                                    + "attempt={}, endpoints={}, clientId={}", topic, messageIds, maxAttempts, attempt,
                            endpoints, clientId);
                }
            }


            //以下就是消息发送失败要执行的回调方法
            @Override
            public void onFailure(Throwable t) {
                //得到发送消息的耗时
                final Duration duration = stopwatch.elapsed();
                //执行消息发送后的回调方法
                doAfter(MessageHookPoints.SEND, messageCommons, duration, MessageHookPointsStatus.ERROR);

                //收集所有发送失败的消息的Id，用于日志记录
                List<MessageId> messageIds = new ArrayList<>();
                //遍历所有发送失败的消息，收集它们的消息Id
                for (PublishingMessageImpl message : messages) {
                    messageIds.add(message.getMessageId());
                }
                //隔离失败的Broker节点，现在大家应该清楚了isolated成员变量中的数据是怎么添加的了吧？
                //消息发送失败则意味着目标Broker节点可能出现了网络故障，那就隔离这个节点
                isolate(endpoints);
                //判断消息重试次数是否达到了最大重试次数，如果达到了最大次数就停止发送消息，记录日志信息
                if (attempt >= maxAttempts) {
                    future.setException(t);
                    LOGGER.error("Failed to send message(s) finally, run out of attempt times, maxAttempts={}, " +
                                    "attempt={}, topic={}, messageId(s)={}, endpoints={}, clientId={}",
                            maxAttempts, attempt, topic, messageIds, endpoints, clientId, t);
                    return;
                }
                //执行到这里意味着没有超过最大重试次数，那么接下来就可以重新发送消息，但在发送消息之前要判断一下
                //当前发送的消息是否为事务消息，如果是事务消息，那就不能重试了，而是要记录错误日志，把异常情况汇报给用户
                if (MessageType.TRANSACTION.equals(messageType)) {
                    future.setException(t);
                    LOGGER.error("Failed to send transactional message finally, maxAttempts=1, attempt={}, " +
                                    "topic={}, messageId(s)={}, endpoints={}, clientId={}", attempt, topic, messageIds,
                            endpoints, clientId, t);
                    return;
                }
                //接下来就是消息重试的操作了，在这里把重试次数加一，方便下一次选取新的目标消息队列
                int nextAttempt = 1 + attempt;
                //判断是否要立刻进行消息重试，TooManyRequestsException是服务端异常，表示服务端正在处理过载请求，不能立即进行消息重试
                if (!(t instanceof TooManyRequestsException)) {
                    LOGGER.warn("Failed to send message, would attempt to resend right now, maxAttempts={}, "
                                    + "attempt={}, topic={}, messageId(s)={}, endpoints={}, clientId={}", maxAttempts, attempt,
                            topic, messageIds, endpoints, clientId, t);
                    //如果不是TooManyRequestsException异常则可以立刻执行消息重试
                    send0(future, topic, messageType, candidates, messages, nextAttempt);
                    return;
                }//如果是TooManyRequestsException异常，那么就要延迟执行消息重试
                //这里得到消息延迟发送时间
                final Duration delay = ProducerImpl.this.getRetryPolicy().getNextAttemptDelay(nextAttempt);
                LOGGER.warn("Failed to send message due to too many requests, would attempt to resend after {}, "
                                + "maxAttempts={}, attempt={}, topic={}, messageId(s)={}, endpoints={}, clientId={}", delay,
                        maxAttempts, attempt, topic, messageIds, endpoints, clientId, t);
                //在这里使用定时任务执行器延迟执行消息重试操作
                clientManager.getScheduler().schedule(() -> send0(future, topic, messageType, candidates, messages,
                        nextAttempt), delay.toNanos(), TimeUnit.NANOSECONDS);
            }
        }, clientCallbackExecutor);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：包装要发送的消息的方法，其实就是把要发送的消息包装到SendMessageRequest请求对象中
     */
    private SendMessageRequest wrapSendMessageRequest(List<PublishingMessageImpl> messages) {
        return SendMessageRequest.newBuilder()
                .addAllMessages(messages.stream().map(PublishingMessageImpl::toProtobuf).collect(Collectors.toList()))
                .build();
    }

    //从负载均衡器中得到消息队列候选列表的方法
    private List<MessageQueueImpl> takeMessageQueues(PublishingLoadBalancer result) {
        return result.takeMessageQueues(isolated, this.getRetryPolicy().getMaxAttempts());
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：得到指定主题的负载均衡器
     */
    private ListenableFuture<PublishingLoadBalancer> getPublishingTopicRouteResult(final String topic) {
        final PublishingLoadBalancer result = publishingRouteDataCache.get(topic);
        if (null != result) {
            return Futures.immediateFuture(result);
        }//如果没有从成员变量中的到，那就创建一个信息返回，同时把新创建的负载均衡器缓存到publishingRouteDataCache成员变量中
        return Futures.transformAsync(getRouteData(topic), topicRouteDataResult -> {
            final PublishingLoadBalancer loadBalancer = new PublishingLoadBalancer(topicRouteDataResult);
            publishingRouteDataCache.put(topic, loadBalancer);
            return Futures.immediateFuture(loadBalancer);
        }, MoreExecutors.directExecutor());
    }
    //生产者客户端向服务端发送事务消息的方法
    @Override
    public SendReceipt send(Message message, Transaction transaction) throws ClientException {

        return null;
    }

    @Override
    public CompletableFuture<SendReceipt> sendAsync(Message message) {
        return null;
    }

    @Override
    public Transaction beginTransaction() {
        return null;
    }

    @Override
    public void close() {
        this.stopAsync().awaitTerminated();
    }

    //该方法在第九版本代码中用不到，等用到的时候，大家就清楚怎么向isolated成员变量中添加数据了
    private void isolate(Endpoints endpoints) {
        isolated.add(endpoints);
    }

    private RetryPolicy getRetryPolicy() {
        return producerSettings.getRetryPolicy();
    }
    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：阻塞线程，等待future操作完成的方法
     */
    protected <T> T handleClientFuture(ListenableFuture<T> future) throws ClientException {
        try {
            return future.get();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            final Throwable cause = e.getCause();
            if (cause instanceof ClientException) {
                throw (ClientException) cause;
            }
            if (cause instanceof RuntimeException) {
                throw (RuntimeException) cause;
            }
            throw new ClientException(null == cause ? e : cause);
        }
    }

    //获取客户端配置信息对象的方法
    public ClientConfiguration getClientConfiguration() {
        return clientConfiguration;
    }
}
